package org.qy.star_server.data.manager;

import lombok.extern.slf4j.Slf4j;
import org.qy.star_common.message.Message;
import org.qy.star_common.message.MessageProperties;
import org.qy.star_common.util.PathUtil;
import org.qy.star_server.core.Queue;
import org.qy.star_server.core.Stat;
import org.qy.star_server.data.center.MemoryDataCenter;
import org.qy.star_server.util.ProtoUtil;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Repository;

import java.io.*;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardCopyOption;
import java.util.LinkedList;
import java.util.List;


/**
 * @author yinger
 * @description 消息存储在文件中 此处使用更加高效的 protobuf
 * 将消息进行封装 进行消息的持久化 存储到文件
 * 此处约定存入文件的消息的协议： 1 字节消息是否有效（0 无效 1 有效） + 4 字节消息长度   + 消息
 * @date 2024/3/26 10:32
 **/
@Slf4j
public class FileManager {
    /**
     * 创建出队列的消息数据文件和消息统计文件
     * 创建队列时，将队列文件创建出来
     *
     * @param queue 队列
     */
    public void createQueueFiles(Queue queue) {
        String dataPath = PathUtil.getQueueMessageDataPath(queue.getUsername(), queue.getVirtualHostName(), queue.getQueueName());
        try {
            // 无论是否存在 都尝试创建目录结构
            PathUtil.createQueueDir(queue.getUsername(), queue.getVirtualHostName(), queue.getQueueName());
            Path path = Paths.get(dataPath);
            if (!Files.exists(path)) {
                Files.createFile(path);
            }
        } catch (IOException e) {
            log.error("创建消息数据文件失败");
            e.printStackTrace();
        }
    }

    /**
     * 删除队列文件
     * 删除队列时，将队列文件进行删除
     *
     * @param queue 队列
     */
    public void deleteQueueFiles(Queue queue) {
        String dataPath = PathUtil.getQueueMessageDataPath(queue.getUsername(), queue.getVirtualHostName(), queue.getQueueName());
        Path path = Paths.get(dataPath);
        if (Files.exists(path)) {
            try {
                Files.delete(path);
            } catch (IOException e) {
                log.error("删除消息数据文件失败");
                e.printStackTrace();
            }
        }
    }

    /**
     * 此处约定调用时，需要保证消息的合法性（消息存在）
     *
     * @param queue   队列
     * @param message 要插入的消息
     */
    public void insertMessage(Queue queue, Message message) {
        queue =  MemoryDataCenter.memory.selectQueue(queue.getUsername(), queue.getVirtualHostName(), queue.getQueueName());
        synchronized (queue) {
            String dataPath = PathUtil.getQueueMessageDataPath(queue.getUsername(), queue.getVirtualHostName(), queue.getQueueName());
            File file = new File(dataPath);

            // 使用 protobuf 将消息序列化
            byte[] body = ProtoUtil.serializeMessage(message);
            message.setOffsetBegin(file.length() + 5);
            message.setOffsetEnd(file.length() + 5 + body.length);
            // TODO 将消息写入文件 进行追加写！！！
            try (OutputStream outputStream = new FileOutputStream(file, true)) {
                try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    // 写入一个字节标志位
                    dataOutputStream.writeByte((byte) 0x1);
                    // 写入4个字节消息长度
                    dataOutputStream.writeInt(body.length);
                    // 写入消息
                    dataOutputStream.write(body);
                    dataOutputStream.flush();
                }
            } catch (IOException e) {
                log.error("写入消息数据文件失败，Queue" + queue + "Message" + message);
                throw new RuntimeException(e);
            }
            // 更新消息统计文件
            Stat stat = queue.getStat();
            int newTotalCount = stat.getFileTotalCount() + 1;
            int newValidCount = stat.getFileValidCount() + 1;
            queue.setFileTotalCount(newTotalCount);
            queue.setFileValidCount(newValidCount);
        }
    }

    /**
     * 删除消息
     *
     * @param queue   队列
     * @param message 消息
     */

    public void deleteMessage(Queue queue, Message message) {
        queue =  MemoryDataCenter.memory.selectQueue(queue.getUsername(), queue.getVirtualHostName(), queue.getQueueName());
        message =  MemoryDataCenter.memory.selectMessageFromCenter(message.getMessageId());
        synchronized (queue) {
            String dataPath = PathUtil.getQueueMessageDataPath(queue.getUsername(), queue.getVirtualHostName(), queue.getQueueName());
            File file = new File(dataPath);
            try (RandomAccessFile randomAccessFile = new RandomAccessFile(file, "rw")) {
                // 计算消息有效标志位的位置
                long offset = message.getOffsetBegin() - 5;
                // 验证offset是否在合理范围内
                if (offset < 0 || offset >= randomAccessFile.length()) {
                    log.error(String.valueOf(offset) + "xxxx");
                    return;
                }
                // 移动到有效标志位
                randomAccessFile.seek(offset);
                byte validityFlag = randomAccessFile.readByte();
                // 如果消息有效，则进行逻辑删除
                if (validityFlag == (byte) 0x1) {
                    randomAccessFile.seek(offset);
                    randomAccessFile.writeByte((byte) 0x0); // 修改有效标志位为无效
                    // 更新统计信息
                    Stat stat = queue.getStat();
                    int newValidCount = stat.getFileValidCount();
                    if (stat.getFileValidCount() > 0) {
                        newValidCount = newValidCount - 1;
                    }
                    queue.setFileValidCount(newValidCount);
                } else {
                    log.error("尝试删除无效消息，Queue: {}, Message: {}", queue, message);
                }

                // 删除消息时，检查当前是否需要进行gc
                if (checkGC(queue)) {
                    gc(queue);
                }

                Stat updatedStat = queue.getStat();
            } catch (FileNotFoundException | IndexOutOfBoundsException e) {
                log.error("文件未找到，无法删除消息，Queue: {}, Message: {}", queue, message, e);
                throw new RuntimeException(e);
            } catch (IOException e) {
                log.error("读写文件时发生错误，无法删除消息，Queue: {}, Message: {}", queue, message, e);
                throw new RuntimeException(e);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }

    /**
     * 加载当前队列的有效消息，只读 ，但是与可能与写操作并发 所以还是加上锁
     *
     * @return 有效消息链表
     */
    public LinkedList<Message> loadValidMessage(Queue queue) throws IOException {
        queue =  MemoryDataCenter.memory.selectQueue(queue.getUsername(), queue.getVirtualHostName(), queue.getQueueName());
        synchronized (queue) {
            LinkedList<Message> validMessages = new LinkedList<>();
            String dataPath = PathUtil.getQueueMessageDataPath(queue.getUsername(), queue.getVirtualHostName(), queue.getQueueName());
            File file = new File(dataPath);
            if (!file.exists()) {
                return validMessages;
            }

            try (InputStream inputStream = new FileInputStream(file);
                 DataInputStream dataInputStream = new DataInputStream(inputStream)) {
                int currentSeek = 0;
                while (dataInputStream.available() > 0) {
                    // 读取有效性标志位
                    byte validityFlag = dataInputStream.readByte();
                    // 读取消息长度
                    int messageLength = dataInputStream.readInt();
                    if (validityFlag != (byte) 0x1) {
                        // 如果标志位表示消息无效，则跳过此条消息
                        dataInputStream.skipBytes(messageLength);
                        currentSeek += (5 + messageLength);
                        continue;
                    }

                    // 分配足够大小的缓冲区以读取消息内容
                    byte[] buffer = new byte[messageLength];
                    dataInputStream.readFully(buffer);

                    // 反序列化消息
                    Message message = ProtoUtil.parseMessage(buffer);

                    // 设置消息在文件中的偏移信息
                    message.setOffsetBegin(currentSeek + 5);
                    message.setOffsetEnd(currentSeek + 5 + messageLength);

                    // 添加有效消息到结果列表
                    validMessages.add(message);
                    currentSeek += (5 + messageLength);
                }
            }
            return validMessages;
        }
    }


    /**
     * 检查是否满足 gc 的条件
     *
     * @param queue 队列
     * @return 是否满足gc条件
     * @throws FileNotFoundException
     */
    public boolean checkGC(Queue queue) throws FileNotFoundException {
        queue =  MemoryDataCenter.memory.selectQueue(queue.getUsername(), queue.getVirtualHostName(), queue.getQueueName());
        synchronized (queue) {
            // 读取队列消息统计文件
            Stat stat = queue.getStat();
            // 此处约定，总消息数量超过 2000 条 && 有效消息数量 低于 50% 时进行队列消息文件的GC
            return stat.getFileTotalCount() > 2000 && (((double) stat.getFileValidCount() < ((double) stat.getFileTotalCount()) * 0.5));
        }
    }

    /**
     * 执行gc 对队列的无效消息进行删除
     * 注意 gc 后消息的偏移值会发生变化，此时应该更新 内存中消息的偏移值
     *
     * @param queue 队列
     * @throws IOException
     * @throws InterruptedException
     */
    public void gc(Queue queue) throws IOException, InterruptedException {

        queue =  MemoryDataCenter.memory.selectQueue(queue.getUsername(), queue.getVirtualHostName(), queue.getQueueName());
        System.out.println("开始执行gc");
        // gc 时，对队列进行加锁
        synchronized (queue) {
            // 获取原始消息文件路径
            String originalFilePath = PathUtil.getQueueMessageDataPath(queue.getUsername(), queue.getVirtualHostName(), queue.getQueueName());

            // 创建临时文件路径用于存储有效消息
            String tempFilePath = PathUtil.getTempQueueMessageDataPath(queue.getUsername(), queue.getVirtualHostName(), queue.getQueueName());

            // 初始化临时文件的偏移量
            long tempFileOffset = 5; // 考虑到一个字节的有效标志位和四个字节的消息长度

            // 加载有效消息并写入临时文件
            LinkedList<Message> validMessages = loadValidMessage(queue);
            try (OutputStream tempOutputStream = new FileOutputStream(tempFilePath);
                 DataOutputStream tempDataOutputStream = new DataOutputStream(tempOutputStream)) {
                for (Message message : validMessages) {
                    // 对于每个有效消息，按照协议格式写入临时文件，并更新其offsetBegin和offsetEnd
                    tempDataOutputStream.writeByte((byte) 0x1); // 写入有效标志位
                    byte[] body = ProtoUtil.serializeMessage(message);
                    tempDataOutputStream.writeInt(body.length); // 写入消息长度

                    message.setOffsetBegin(tempFileOffset);
                    message.setOffsetEnd(tempFileOffset + body.length);

                    // 写入消息本体
                    tempDataOutputStream.write(body);
                    tempFileOffset += 5 + body.length;
                }
                tempDataOutputStream.flush();
            }

            // 删除原始消息文件
            File originalFile = new File(originalFilePath);
            boolean isDeleted = originalFile.delete();
            if (!isDeleted) {
                log.error("无法删除原始消息文件: {}", originalFilePath);
                throw new IOException("无法删除原始消息文件");
            }

            // 将临时文件重命名至原始消息文件路径，此处这个方法可以保证消息的原子性
            Files.move(Paths.get(tempFilePath), Paths.get(originalFilePath), StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING);

            // 刷新内存中消息的偏移值
            updateQueueMessageOffset(queue.getUsername(),queue.getVirtualHostName(),queue.getQueueName());
            // 更新消息统计文件（因为GC可能会导致有效消息数量变化）
            queue.setFileTotalCount(validMessages.size());
            queue.setFileValidCount(validMessages.size());
            log.info("已完成对Queue {} 的垃圾回收", queue.getQueueName());
        }
    }

    public LinkedList<Message> loadValidMessages(String username, String virtualHostName, String queueName) throws IOException {
        return loadValidMessage(Queue.getQueryQueue(username, virtualHostName, queueName));
    }
    public void updateQueueMessageOffset(String username, String virtualHostName, String queueName) {
        try {
            Queue.QueueGloballyUniqueIdentity queueGloballyUniqueIdentity = new Queue.QueueGloballyUniqueIdentity(username, virtualHostName, queueName);
            List<Message> messages = loadValidMessages(username, virtualHostName, queueName);
            // 释放内存中的消息数据重新插入
            MemoryDataCenter.memory.getMessageCenter().clear();
            for (Message message : messages) {
                MemoryDataCenter.memory.getMessageCenter().put(message.getMessageId(), message);
            }
            MemoryDataCenter.memory.getQueueMessageCenter().put(queueGloballyUniqueIdentity, new LinkedList<>(messages));
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public static void testInsertMessage(Queue queue) {
        new FileManager().createQueueFiles(queue);
        for (int i = 0; i < 500; i++) {
            new FileManager().insertMessage(Queue.getQueryQueue("admin", "star", "testQueue"), Message.createMessageWithId(new MessageProperties("", "testQueue", 2), "hello".getBytes(), "123456"));
        }
    }


    public static void main(String[] args) throws IOException {
        Queue queue = Queue.getQueryQueue("admin", "star", "testQueue");
        testInsertMessage(queue);


    }


}
